package com.inventive.jmscache.server.hazelcast;

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.IMap;
import com.hazelcast.core.IQueue;
import com.inventive.jmscache.server.BackendTopic;
import com.inventive.jmscache.server.JmsCacheException;
import com.inventive.jmscache.server.JmsCacheMessage;

import java.util.concurrent.TimeUnit;

/**
 * @author Grant Little <grant.little@coreservices.com.au>
 */
public class HazelcastCachedTopic implements BackendTopic {

    private String name;

    public HazelcastCachedTopic(String name) {
        this.name = name;
    }

    public void publish(String messageId, JmsCacheMessage message) {
        Hazelcast.getMap(name).put(messageId, message);
    }

    public JmsCacheMessage poll(String subscriptionId) {
        //This is not the most efficient way of doing it. Needs to be refactored
        IQueue<String> queue = Hazelcast.getQueue(subscriptionId);
        String messageId = queue.poll();
        IMap<String, JmsCacheMessage> messagesMap = Hazelcast.getMap(name);
        return messagesMap.get(messageId);
    }

    public JmsCacheMessage poll(String subscriptionId, long timeoutMillis) {
        //This is not the most efficient way of doing it. Needs to be refactored
        IQueue<String> queue = Hazelcast.getQueue(subscriptionId);
        try {
            String messageId = queue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
            IMap<String, JmsCacheMessage> messagesMap = Hazelcast.getMap(name);
            return messagesMap.get(messageId);
        } catch (InterruptedException e) {
            throw new JmsCacheException(e);
        }
    }

}
